/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */


package com.sui.bigdata.flink.sql.core.parser;

import com.sui.bigdata.flink.sql.core.util.DtStringUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.shaded.curator.org.apache.curator.shaded.com.google.common.collect.Lists;
import org.apache.flink.shaded.guava18.com.google.common.base.Strings;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * Reason:
 * Date: 2019/07/10
 *
 * @author yong_chen@sui.com
 */

public class MetaParser {

    private static final char SQL_DELIMITER = ';';
    private static org.apache.log4j.Logger logger = org.apache.log4j.Logger.getLogger(MetaParser.class);


    private static List<IParser> sqlParserList = Lists.newArrayList(CreateTableParser.newInstance());


    /**
     * flink support sql syntax
     * CREATE TABLE sls_stream() with ();
     * CREATE (TABLE|SCALA) FUNCTION fcnName WITH com.dtstack.com;
     * insert into tb1 select * from tb2;
     *
     * @param sql
     */
    public static List<MetaParserResult> parseSql(String sql, Map<String, String> basicProp) throws Exception {

        if (StringUtils.isBlank(sql)) {
            throw new RuntimeException("sql is not null");
        }


        sql = sql.replaceAll("--.*", "")
                .replaceAll("\r\n", " ")
                .replaceAll("\n", " ")
                .replace("\t", " ").trim();

        List<String> sqlArr = DtStringUtil.splitIgnoreQuota(sql, SQL_DELIMITER);
        SqlTree sqlTree = new SqlTree();
        for (String childSql : sqlArr) {
            if (Strings.isNullOrEmpty(childSql)) {
                continue;
            }
            boolean result = false;
            for (IParser sqlParser : sqlParserList) {
                if (!sqlParser.verify(childSql)) {
                    continue;
                }

                sqlParser.parseSql(childSql, sqlTree);
                result = true;
            }

            if (!result) {
                throw new RuntimeException("Syntax does not support.");
            }
        }

        List<MetaParserResult> metaParserResults = new ArrayList();
        for (Map.Entry<String, CreateTableParser.SqlParserResult> result : sqlTree.getPreDealTableMap().entrySet()) {
            MetaParserResult metaParserResult = new MetaParserResult();
            Map<String, Object> prop = result.getValue().getPropMap();
            fillBasicProp(prop,basicProp);
            metaParserResult.setTableName(result.getValue().getTableName());


            String[] arr = String.valueOf(prop.get("type")).split("\\.");
            if (arr.length !=2){
                 metaParserResult.setErrorMsg(" type should be like  source.kafka");
            }else {
                metaParserResult.setType(arr[0]);
                metaParserResult.setComponent(arr[1]);
            }

            metaParserResult.setFieldInfo(result.getValue().getFieldsInfoStr());
            prop.remove("type");
            metaParserResult.setPropMap(prop);
            metaParserResults.add(metaParserResult);
        }
        return metaParserResults;
    }

    public static void fillBasicProp(Map<String, Object> prop, Map<String, String> basicInfo) {
        String type = String.valueOf(prop.get("type"));
        switch (type.split("\\.")[type.split("\\.").length-1]){
            case "kafka":
                putProp("bootstrapServers",prop,basicInfo);
                break;
            case "hbase":
                putProp("zookeeperQuorum",prop,basicInfo);
                break;
            case "mysql":
                putProp("url",prop,basicInfo);
                putProp("userName",prop,basicInfo);
                putProp("password",prop,basicInfo);
                break;
            case "kudu":
                putProp("kuduMasters",prop,basicInfo);
            default:
                return;
        }
    }

    public static void putProp(String propName,Map<String, Object> prop,Map<String, String> basicInfo){
        if ( null == prop.get(propName)){
            prop.put(propName,basicInfo.get(propName));
        }
    }

    public static class MetaParserResult {

        private String tableName;

        private String type;

        private String component;

        private String fieldInfo;

        private Map<String, Object> propMap;

        private String errorMsg;

        public String getType() {
            return type;
        }

        public void setType(String type) {
            this.type = type;
        }

        public String getComponent() {return component;}

        public void setComponent(String component) {
            this.component = component;
        }

        public Map<String, Object> getPropMap() {
            return propMap;
        }

        public void setPropMap(Map<String, Object> propMap) {
            this.propMap = propMap;
        }

        public String getTableName() {
            return tableName;
        }

        public void setTableName(String tableName) {
            this.tableName = tableName;
        }

        public String getFieldInfo() {
            return fieldInfo;
        }

        public void setFieldInfo(String fieldInfo) {
            this.fieldInfo = fieldInfo;
        }

        public String getErrorMsg() {
            return errorMsg;
        }

        public void setErrorMsg(String errorMsg) {
            this.errorMsg = errorMsg;
        }
    }


}
